package org.lemon.transport.coap.client;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.californium.core.coap.CoAP;
import org.eclipse.californium.core.coap.Response;
import org.eclipse.californium.core.server.resources.CoapExchange;
import org.lemon.transport.coap.server.service.callback.CoapNoOpCallback;
import org.lemon.transport.context.service.TransportDeviceProfileCache;
import org.lemon.transport.context.adaptor.AdaptorException;
import org.lemon.transport.context.auth.SessionInfoCreator;
import org.lemon.transport.context.auth.ValidateDeviceCredentialsResponse;
import org.lemon.transport.coap.adapter.CoapTransportAdaptor;
import org.lemon.transport.coap.server.service.CoapMessageObserver;
import org.lemon.transport.coap.server.service.CoapService;
import org.lemon.transport.coap.server.service.CoapTransportContext;
import org.lemon.transport.coap.server.service.TransportConfigurationContainer;
import org.lemon.transport.coap.server.service.callback.AbstractSyncSessionCallback;
import org.lemon.transport.context.data.device.*;
import org.lemon.transport.context.data.device.profile.*;
import org.lemon.transport.context.service.TransportService;
import org.server.transport.gen.SessionProto;
import org.server.transport.gen.TransportProto;
import org.lemon.transport.context.service.TransportServiceCallback;
import org.lemon.transport.context.service.event.DeviceDeletedEvent;
import org.lemon.transport.context.service.event.DeviceProfileUpdatedEvent;
import org.lemon.transport.context.service.event.DeviceUpdatedEvent;
import org.lemon.transport.context.session.SessionMsgListener;
import org.lemon.transport.context.session.SessionMsgType;
import org.springframework.context.annotation.Lazy;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Service;

import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.*;

import static org.eclipse.californium.core.coap.Message.MAX_MID;
import static org.eclipse.californium.core.coap.Message.NONE;

@Slf4j
@Service
public class CoapClientContext {

    private final CoapService coapService;
    private final CoapTransportContext transportContext;
    private final TransportService transportService;
    private final TransportDeviceProfileCache profileCache;

    //<设备id, CoapClientState>
    private final ConcurrentMap<Long, CoapClientState> clients = new ConcurrentHashMap<>();
    private final ConcurrentMap<String, CoapClientState> clientsByToken = new ConcurrentHashMap<>();

    public CoapClientContext(CoapService coapService, @Lazy CoapTransportContext transportContext, TransportService transportService) {
        this.coapService = coapService;
        this.transportContext = transportContext;
        this.transportService = transportService;
        this.profileCache = transportContext.getTransportDeviceProfileCache();
    }

    @EventListener(DeviceProfileUpdatedEvent.class)
    public void onApplicationEvent(DeviceProfileUpdatedEvent event) {

        DeviceProfile deviceProfile = event.getDeviceProfile();
        clients.values().stream().filter(state -> state.getSession() == null).forEach(state -> {
            state.lock();
            try {
                if (deviceProfile.getId().equals(state.getProfileId())) {
                    initStateAdaptor(deviceProfile, state);
                }
            } catch (AdaptorException e) {
                log.trace("[{}] Failed to update client state due to: ", state.getDeviceId(), e);
            } finally {
                state.unlock();
            }
        });
    }

    @EventListener(DeviceUpdatedEvent.class)
    public void onApplicationEvent(DeviceUpdatedEvent event) {

        Device device = event.getDevice();
        CoapClientState state = clients.get(device.getDeviceId());
        if (state == null) {
            return;
        }
        state.lock();
        try {
            if (state.getSession() == null) {
                clients.remove(device.getDeviceId());
            }
        } finally {
            state.unlock();
        }
    }

    @EventListener(DeviceDeletedEvent.class)
    public void onApplicationEvent(DeviceDeletedEvent event) {

        clients.remove(event.getDeviceId());
    }

    public boolean registerAttributeObservation(CoapClientState clientState, String token, CoapExchange exchange) {

        clientState.lock();
        try {
            boolean newObservation;
            if (clientState.getAttrsObservationState() == null) {
                newObservation = true;
                clientState.setAttrsObservationState(new CoapObservationState(exchange, token));
            } else {
                newObservation = clientState.getAttrsObservationState().getToken().equals(token);
                if (newObservation) {
                    CoapObservationState old = clientState.getAttrsObservationState();
                    clientState.setAttrsObservationState(new CoapObservationState(exchange, token));
                    old.getExchange().respond(CoAP.ResponseCode.DELETED);
                }
            }
            if (newObservation) {
                clientsByToken.put(token, clientState);
                if (clientState.getSession() == null) {
                    registerNewSession(clientState);
                }
                TransportProto.SubscribeToAttributeUpdatesMsg.Builder subscribeUpdateMsgBuilder = TransportProto.SubscribeToAttributeUpdatesMsg.newBuilder();
                subscribeUpdateMsgBuilder.setSubscribe(true);
                subscribeUpdateMsgBuilder.setSessionType(SessionProto.SessionType.SYNC);
                transportService.process(clientState.getSession(), subscribeUpdateMsgBuilder.build(), new CoapNoOpCallback(exchange));

//                TransportProto.GetAttributeRequestMsg.Builder getAttributeMsgBuilder = TransportProto.GetAttributeRequestMsg.newBuilder();
//                getAttributeMsgBuilder.setOnlyShared(true);
//                transportService.process(clientState.getSession(), getAttributeMsgBuilder.build(), new CoapNoOpCallback(exchange));
            }
            return newObservation;
        } finally {
            clientState.unlock();
        }
    }

    public void deregisterAttributeObservation(CoapClientState clientState, String token, CoapExchange exchange) {

        clientState.lock();
        try {
            clientsByToken.remove(token);
            if (clientState.getSession() == null) {
                log.trace("[{}] Failed to delete attribute observation: {}. Session is not present.", clientState.getDeviceId(), token);
                return;
            }
            if (clientState.getAttrsObservationState() == null) {
                log.trace("[{}] Failed to delete attribute observation: {}. It is not registered.", clientState.getDeviceId(), token);
                return;
            }
            if (!clientState.getAttrsObservationState().getToken().equals(token)) {
                log.trace("[{}] Failed to delete attribute observation: {}. Token mismatch.", clientState.getDeviceId(), token);
                return;
            }
            cancelAttributeSubscription(clientState);
        } finally {
            clientState.unlock();
        }
    }

    public boolean registerRpcObservation(CoapClientState clientState, String token, CoapExchange exchange) {
        clientState.lock();
        try {
            boolean newObservation;
            if (clientState.getRpcObservationState() == null) {
                newObservation = true;
                clientState.setRpcObservationState(new CoapObservationState(exchange, token));
            } else {
                newObservation = clientState.getRpcObservationState().getToken().equals(token);
                if (newObservation) {
                    CoapObservationState old = clientState.getRpcObservationState();
                    clientState.setRpcObservationState(new CoapObservationState(exchange, token));
                    old.getExchange().respond(CoAP.ResponseCode.DELETED);
                }
            }
            if (newObservation) {
                clientsByToken.put(token, clientState);
                if (clientState.getSession() == null) {
                    registerNewSession(clientState);
                }
                TransportProto.SubscribeToRPCMsg.Builder subscribeRpcMsgBuilder = TransportProto.SubscribeToRPCMsg.newBuilder();
                subscribeRpcMsgBuilder.setSubscribe(true);
                subscribeRpcMsgBuilder.setSessionType(SessionProto.SessionType.SYNC);
                transportService.process(clientState.getSession(), subscribeRpcMsgBuilder.build(), new CoapNoOpCallback(exchange));
            }
            return newObservation;
        } finally {
            clientState.unlock();
        }
    }

    public void deregisterRpcObservation(CoapClientState clientState, String token, CoapExchange exchange) {

        clientState.lock();
        try {
            clientsByToken.remove(token);
            if (clientState.getSession() == null) {
                log.trace("[{}] Failed to delete attribute observation: {}. Session is not present.", clientState.getDeviceId(), token);
                return;
            }
            if (clientState.getRpcObservationState() == null) {
                log.trace("[{}] Failed to delete rpc observation: {}. It is not registered.", clientState.getDeviceId(), token);
                return;
            }
            if (!clientState.getRpcObservationState().getToken().equals(token)) {
                log.trace("[{}] Failed to delete rpc observation: {}. Token mismatch.", clientState.getDeviceId(), token);
                return;
            }
            cancelAttributeSubscription(clientState);
        } finally {
            clientState.unlock();
        }
    }

    public SessionProto.SessionInfoProto getNewSession(CoapClientState state) {
        return SessionInfoCreator.create(state.getCredentials(), transportContext, UUID.randomUUID());
    }

    private void registerNewSession(CoapClientState clientState) {
        SessionProto.SessionInfoProto session = SessionInfoCreator.create(clientState.getCredentials(), transportContext, UUID.randomUUID());
        clientState.setSession(session);
        CoapSessionListener listener = new CoapSessionListener(clientState);
        clientState.setListener(listener);
        transportService.registerAsyncSession(session, clientState.getListener());
        transportService.process(session, getSessionEventMsg(SessionProto.SessionEvent.OPEN), null);
    }

    private void initStateAdaptor(DeviceProfile deviceProfile, CoapClientState state) throws AdaptorException {

        state.setConfiguration(getTransportConfigurationContainer(deviceProfile));
        state.setAdaptor(getCoapTransportAdaptor(state.getConfiguration().isJsonPayload()));
        state.setContentFormat(state.getAdaptor().getContentFormat());
    }

    private TransportConfigurationContainer getTransportConfigurationContainer(DeviceProfile deviceProfile) throws AdaptorException {

        DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration();
        if (transportConfiguration instanceof DefaultDeviceProfileTransportConfiguration) {
            return new TransportConfigurationContainer(true);
        } else if (transportConfiguration instanceof CoapDeviceProfileTransportConfiguration) {
            CoapDeviceProfileTransportConfiguration coapDeviceProfileTransportConfiguration =
                    (CoapDeviceProfileTransportConfiguration) transportConfiguration;
            CoapDeviceTypeConfiguration coapDeviceTypeConfiguration =
                    coapDeviceProfileTransportConfiguration.getCoapDeviceTypeConfiguration();
            if (coapDeviceTypeConfiguration instanceof DefaultCoapDeviceTypeConfiguration) {
                DefaultCoapDeviceTypeConfiguration defaultCoapDeviceTypeConfiguration =
                        (DefaultCoapDeviceTypeConfiguration) coapDeviceTypeConfiguration;
                TransportPayloadTypeConfiguration transportPayloadTypeConfiguration =
                        defaultCoapDeviceTypeConfiguration.getTransportPayloadTypeConfiguration();
                if (transportPayloadTypeConfiguration instanceof JsonTransportPayloadConfiguration) {
                    return new TransportConfigurationContainer(true);
                } else {
                    ProtoTransportPayloadConfiguration protoTransportPayloadConfiguration =
                            (ProtoTransportPayloadConfiguration) transportPayloadTypeConfiguration;
                    String deviceTelemetryProtoSchema = protoTransportPayloadConfiguration.getDeviceTelemetryProtoSchema();
                    String deviceAttributesProtoSchema = protoTransportPayloadConfiguration.getDeviceAttributesProtoSchema();
                    String deviceRpcRequestProtoSchema = protoTransportPayloadConfiguration.getDeviceRpcRequestProtoSchema();
                    String deviceRpcResponseProtoSchema = protoTransportPayloadConfiguration.getDeviceRpcResponseProtoSchema();
                    return new TransportConfigurationContainer(false,
                            protoTransportPayloadConfiguration.getTelemetryDynamicMessageDescriptor(deviceTelemetryProtoSchema),
                            protoTransportPayloadConfiguration.getAttributesDynamicMessageDescriptor(deviceAttributesProtoSchema),
                            protoTransportPayloadConfiguration.getRpcResponseDynamicMessageDescriptor(deviceRpcResponseProtoSchema),
                            protoTransportPayloadConfiguration.getRpcRequestDynamicMessageBuilder(deviceRpcRequestProtoSchema)
                    );
                }
            } else {
                throw new AdaptorException("Invalid CoapDeviceTypeConfiguration type: " + coapDeviceTypeConfiguration.getClass().getSimpleName() + "!");
            }
        } else {
            throw new AdaptorException("Invalid DeviceProfileTransportConfiguration type" + transportConfiguration.getClass().getSimpleName() + "!");
        }
    }

    private CoapTransportAdaptor getCoapTransportAdaptor(boolean jsonPayloadType) {
        return jsonPayloadType ? transportContext.getJsonCoapAdaptor() : transportContext.getProtoCoapAdaptor();
    }

    private void cancelRpcSubscription(CoapClientState state) {
        if (state.getRpcObservationState() != null) {
            clientsByToken.remove(state.getRpcObservationState().getToken());
            CoapExchange exchange = state.getRpcObservationState().getExchange();
            state.setRpcObservationState(null);
            transportService.process(state.getSession(), TransportProto.SubscribeToRPCMsg.newBuilder().setSubscribe(false).build(),
                    new TransportServiceCallback<Void>() {
                        @Override
                        public void onSuccess(Void msg) {
                            exchange.respond(new Response(CoAP.ResponseCode.DELETED));
                        }

                        @Override
                        public void onError(Throwable e) {
                            exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR);
                        }
                    });
            if (state.getAttrsObservationState() == null) {
                closeAndCleanup(state);
            }
        }
    }

    private void cancelAttributeSubscription(CoapClientState state) {

        if (state.getAttrsObservationState() != null) {
            clientsByToken.remove(state.getAttrsObservationState().getToken());
            CoapExchange exchange = state.getRpcObservationState().getExchange();
            state.setAttrsObservationState(null);
            transportService.process(state.getSession(), TransportProto.SubscribeToAttributeUpdatesMsg.newBuilder().setSubscribe(false).build(),
                    new TransportServiceCallback<Void>() {
                        @Override
                        public void onSuccess(Void msg) {
                            exchange.respond(new Response(CoAP.ResponseCode.DELETED));
                        }

                        @Override
                        public void onError(Throwable e) {
                            exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR);
                        }
                    });
        }
    }

    private void closeAndCleanup(CoapClientState state) {

        transportService.process(state.getSession(), getSessionEventMsg(SessionProto.SessionEvent.CLOSED), null);
        transportService.deregisterSession(state.getSession());
        state.setSession(null);
        state.setConfiguration(null);
        state.setCredentials(null);
        state.setAdaptor(null);
    }

    /**
     * 是否允许下行数据
     *
     * @param clientState
     * @return
     */
    private boolean isDownlinkAllowed(CoapClientState clientState) {

        PowerMode powerMode = clientState.getPowerMode();
        PowerSavingConfiguration powerSavingConfiguration = null;
        if (powerMode == null) {
            Optional<CoapDeviceProfileTransportConfiguration> clientProfile = getTransportConfiguration(clientState.getProfileId());
            if (clientProfile.isPresent()) {
                powerSavingConfiguration = clientProfile.get().getClientSettings();
                if (powerSavingConfiguration != null) {
                    powerMode = powerSavingConfiguration.getPowerMode();
                }
            }
        }
        if (powerMode == null || PowerMode.DRX.equals(powerMode)) {
            return true;
        }
        clientState.lock();
        long timeSinceLastUplink = System.currentTimeMillis() - clientState.getLastUplinkTime();
        try {
            if (PowerMode.PSM.equals(powerMode)) {
                Long psmActivityTimer = clientState.getPsmActivityTimer();
                if (psmActivityTimer == null && powerSavingConfiguration != null) {
                    psmActivityTimer = powerSavingConfiguration.getPsmActivityTimer();
                }
                if (psmActivityTimer == null || psmActivityTimer == 0L) {
                    psmActivityTimer = coapService.getPsmActivityTimer();
                }
                return timeSinceLastUplink <= psmActivityTimer;
            } else {
                Long pagingTransmissionWindow = clientState.getPagingTransmissionWindow();
                if (pagingTransmissionWindow == null && powerSavingConfiguration != null) {
                    pagingTransmissionWindow = powerSavingConfiguration.getPagingTransmissionWindow();
                }
                if (pagingTransmissionWindow == null || pagingTransmissionWindow == 0L) {
                    pagingTransmissionWindow = coapService.getPagingTransmissionWindow();
                }
                boolean allowed = timeSinceLastUplink <= pagingTransmissionWindow;
                if (!allowed) {
                    return clientState.checkFirstDownlink();
                } else {
                    return true;
                }
            }
        } finally {
            clientState.unlock();
        }
    }

    private Optional<CoapDeviceProfileTransportConfiguration> getTransportConfiguration(Long profileId) {

        DeviceProfile deviceProfile = profileCache.get(profileId);
        if (deviceProfile.getTransportType().equals(DeviceTransportType.COAP)) {
            DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration();
            return Optional.of((CoapDeviceProfileTransportConfiguration) transportConfiguration);
        } else if (deviceProfile.getTransportType().equals(DeviceTransportType.DEFAULT)) {
            return Optional.empty();
        } else {
            log.warn("[{}] Invalid device profile type: {}", profileId, deviceProfile.getTransportType());
            throw new IllegalArgumentException("Invalid device profile type: " + deviceProfile.getTransportType());
        }
    }

    private static SessionProto.SessionEventMsg getSessionEventMsg(SessionProto.SessionEvent event) {
        return SessionProto.SessionEventMsg.newBuilder()
                .setSessionType(SessionProto.SessionType.ASYNC)
                .setEvent(event).build();
    }

    /**
     * 客户端是否sleep的
     *
     * @param client
     * @return
     */
    private boolean asleep(CoapClientState client) {
        boolean changed = compareAndSetSleepFlag(client, true);
        if (changed) {
            log.debug("[{}] client is sleeping", client.getDeviceId());
            transportService.log(client.getSession(), "Info: Client is sleeping!");
        }
        return changed;
    }

    /**
     * 唤醒客户端
     *
     * @param client
     * @return
     */
    public boolean awake(CoapClientState client) {
        return awake(client, true, System.currentTimeMillis());
    }

    private boolean awake(CoapClientState client, boolean notifyOtherServers, long upLinkTs) {
        onUplink(client, notifyOtherServers, upLinkTs);
        boolean changed = compareAndSetSleepFlag(client, false);
        if (changed) {
            log.debug("[{}] client is awake", client.getDeviceId());
            transportService.log(client.getSession(), "Info: Client is awake!");
            sendMessagesAfterSleeping(client);
        }
        return changed;
    }

    private void sendMessagesAfterSleeping(CoapClientState client) {

        if (client.getRpcObservationState() != null) {

        }
        if (client.getAttrsObservationState() != null && client.getMissedAttributeUpdates() != null) {
            client.getListener().onAttributeUpdate(new UUID(client.getSession().getSessionIdMSB(), client.getSession().getSessionIdLSB()),
                    client.getMissedAttributeUpdates());
        }
    }

    /**
     * 上行
     *
     * @param client
     * @param notifyOtherServers
     * @param upLinkTs
     */
    private void onUplink(CoapClientState client, boolean notifyOtherServers, long upLinkTs) {

        PowerMode powerMode = client.getPowerMode();
        PowerSavingConfiguration powerSavingConfiguration = null;
        if (powerMode == null) {
            Optional<CoapDeviceProfileTransportConfiguration> transportConfiguration = getTransportConfiguration(client.getProfileId());
            if (transportConfiguration.isPresent()) {
                powerSavingConfiguration = transportConfiguration.get().getClientSettings();
                if (powerSavingConfiguration != null) {
                    powerMode = powerSavingConfiguration.getPowerMode();
                }
            }
        }
        if (powerMode == null || PowerMode.DRX.equals(powerMode)) {
            client.updateLastUplinkTime(upLinkTs);
            return;
        }
        client.lock();
        try {
            long uplinkTime = client.updateLastUplinkTime(upLinkTs);
            long timeout = getTimeout(client, powerMode, powerSavingConfiguration);
            Future<Void> sleepTask = client.getSleepTask();
            if (sleepTask != null) {
                sleepTask.cancel(false);
            }
            transportContext.getScheduler().schedule(() -> {
                if (uplinkTime == client.getLastUplinkTime()) {
                    asleep(client);
                }
                return null;
            }, timeout, TimeUnit.MILLISECONDS);
            client.setSleepTask(sleepTask);
            //TODO 是否通知其他服务
        } finally {
            client.unlock();
        }
    }

    private long getTimeout(CoapClientState client, PowerMode powerMode, PowerSavingConfiguration powerSavingConfiguration) {

        long timeout;
        if (PowerMode.PSM.equals(powerMode)) {
            Long psmActivityTimer = client.getPsmActivityTimer();
            if (psmActivityTimer == null && powerSavingConfiguration != null) {
                psmActivityTimer = powerSavingConfiguration.getPsmActivityTimer();
            }
            if (psmActivityTimer == null || psmActivityTimer == 0L) {
                psmActivityTimer = coapService.getPsmActivityTimer();
            }
            timeout = psmActivityTimer;
        } else {
            Long pagingTransmissionWindow = client.getPagingTransmissionWindow();
            if (pagingTransmissionWindow == null && powerSavingConfiguration != null) {
                pagingTransmissionWindow = powerSavingConfiguration.getPagingTransmissionWindow();

            }
            if (pagingTransmissionWindow == null || pagingTransmissionWindow == 0L) {
                pagingTransmissionWindow = coapService.getPagingTransmissionWindow();
            }
            timeout = pagingTransmissionWindow;
        }
        return timeout;
    }

    private boolean compareAndSetSleepFlag(CoapClientState client, boolean sleeping) {

        if (sleeping == client.isAsleep()) {
            log.trace("[{}] Client is already at sleeping: {}, ignoring event: {}", client.getDeviceId(), client.isAsleep(), sleeping);
            return false;
        }
        client.lock();
        try {
            if (sleeping == client.isAsleep()) {
                log.trace("[{}] Client is already at sleeping: {}, ignoring event: {}", client.getDeviceId(), client.isAsleep(), sleeping);
                return false;
            }
            PowerMode powerMode = getPowerMode(client);
            if (PowerMode.PSM.equals(powerMode) || PowerMode.E_DRX.equals(powerMode)) {
                log.trace("[{}] Switch sleeping from: {} to: {}", client.getDeviceId(), client.isAsleep(), sleeping);
                client.setAsleep(sleeping);
                return true;
            } else {
                return false;
            }
        } finally {
            client.unlock();
        }
    }

    protected int getNextMsgId() {
        return ThreadLocalRandom.current().nextInt(NONE, MAX_MID + 1);
    }

    private PowerMode getPowerMode(CoapClientState client) {
        PowerMode powerMode = client.getPowerMode();
        if (powerMode == null) {
            Optional<CoapDeviceProfileTransportConfiguration> deviceProfile = getTransportConfiguration(client.getProfileId());
            if (deviceProfile.isPresent()) {
                powerMode = deviceProfile.get().getClientSettings().getPowerMode();
            } else {
                powerMode = PowerMode.PSM;
            }
        }
        return powerMode;
    }

    private void respond(CoapExchange exchange, Response response, int defContentFormat) {
        response.getOptions().setContentFormat(CoapContentFormatUtil.getContentFormat(exchange.getRequestOptions().getContentFormat(), defContentFormat));
        exchange.respond(response);
    }

    public CoapClientState getOrCreateClient(SessionMsgType type, ValidateDeviceCredentialsResponse deviceCredentials, DeviceProfile deviceProfile) throws AdaptorException {

        Long deviceId = deviceCredentials.getDeviceInfo().getDeviceId();
        CoapClientState clientState = clients.computeIfAbsent(deviceId, CoapClientState::new);
        clientState.lock();
        try {
            if (clientState.getConfiguration() == null || clientState.getAdaptor() == null) {
                initStateAdaptor(deviceProfile, clientState);
            }
            if (clientState.getCredentials() == null) {
                clientState.init(deviceCredentials);
            }
        } finally {
            clientState.unlock();
        }
        return clientState;
    }

    /**
     * coap消息订阅监听器
     */
    public class CoapSessionListener implements SessionMsgListener {

        private final CoapClientState state;

        public CoapSessionListener(CoapClientState state) {
            this.state = state;
        }

        @Override
        public void onRemoteSessionCloseCommand(UUID sessionId, TransportProto.SessionCloseNotificationProto sessionCloseNotification) {

        }

        @Override
        public void onGetAttributesResponse(TransportProto.GetAttributeResponseMsg attributesResponse) {

        }

        @Override
        public void onAttributeUpdate(UUID sessionId, TransportProto.AttributeUpdateNotificationMsg attributeUpdateNotification) {

            if (!isDownlinkAllowed(state)) {
                log.trace("[{}] ignore downLink request cause client is sleeping.", state.getDeviceId());
                state.lock();
                try {
                    state.addQueuedNotification(attributeUpdateNotification);
                } finally {
                    state.unlock();
                }
                return;
            }
            log.trace("[{}] Received attributes update notification to device", sessionId);
            CoapObservationState attrs = state.getAttrsObservationState();
            if (attrs != null) {
                try {
                    boolean conRequest = AbstractSyncSessionCallback.isConRequest(state.getAttrsObservationState());
                    int requestId = getNextMsgId();
                    Response response = state.getAdaptor().convertToPublish(attributeUpdateNotification);
                    response.setConfirmable(conRequest);
                    response.setMID(requestId);
                    if (conRequest) {
                        response.addMessageObserver(new CoapMessageObserver(requestId, id -> awake(state), id -> asleep(state)));
                    }
                    respond(attrs.getExchange(), response, state.getContentFormat());
                } catch (AdaptorException e) {
                    log.trace("[{}] Failed to reply due to error", state.getDeviceId(), e);
                    cancelObserveRelation(attrs);
                    cancelAttributeSubscription(state);
                }
            } else {
                log.debug("[{}] Get Attrs exchange is empty", state.getDeviceId());
            }
        }

        @Override
        public void onDeviceUpdate(SessionProto.SessionInfoProto sessionInfo, Device device, Optional<DeviceProfile> deviceProfileOpt) {

            if (deviceProfileOpt.isPresent()) {
                try {
                    initStateAdaptor(deviceProfileOpt.get(), state);
                } catch (AdaptorException e) {
                    log.warn("[{}] Failed to update device: ", device, e);
                }
            }
            state.onDeviceUpdate(device);
        }

        @Override
        public void onDeviceDeleted(Long deviceId) {
            cancelRpcSubscription(state);
            cancelAttributeSubscription(state);
        }

        @Override
        public void onDeviceProfileUpdate(SessionProto.SessionInfoProto newSessionInfo, DeviceProfile deviceProfile) {

            try {
                initStateAdaptor(deviceProfile, state);
            } catch (AdaptorException e) {
                log.warn("[{}] Failed to update device profile: ", deviceProfile.getId(), e);
            }
        }

        @Override
        public void onToDeviceRpcRequest(UUID sessionId, TransportProto.ToDeviceRpcRequestMsg toDeviceRequest) {

        }

        @Override
        public void onToServerRpcResponse(TransportProto.ToServerRpcResponseMsg toServerResponse) {

        }

        private void initStateAdaptor(DeviceProfile deviceProfile, CoapClientState state) throws AdaptorException {
            state.setConfiguration(getTransportConfigurationContainer(deviceProfile));
            state.setAdaptor(getCoapTransportAdaptor(state.getConfiguration().isJsonPayload()));
            state.setContentFormat(state.getAdaptor().getContentFormat());
        }

        private void cancelObserveRelation(CoapObservationState attrs) {
            if (attrs.getObserveRelation() != null) {
                attrs.getObserveRelation().cancel();
            }
        }
    }
}
